package drds.server.execute_engine.data_handling;

import java.nio.charset.StandardCharsets;
import java.util.ArrayList;
import java.util.Arrays;
import java.util.Collections;
import java.util.Iterator;
import java.util.List;

import drds.server.net.packet.RowDataPacket;
import drds.server.util.ByteUtil;
import drds.server.util.CompareUtil;
import drds.server.util.LongUtil;

/**
 * implement group function select a,count(*),sum(*) from A group by a
 */
public class RowDataPacketGrouper {

	private List<RowDataPacket> rowDataPacketList = Collections.synchronizedList(new ArrayList<RowDataPacket>());
	private final AggregateFunctionType[] aggregateFunctionTypes;
	private final int[] groupByColumnIndexs;
	private boolean isMergAvg = false;
	private GroupByFilter groupByFilter;

	public RowDataPacketGrouper(int[] groupByColumnIndexs, AggregateFunctionType[] aggregateFunctionTypes, GroupByFilter groupByFilter) {
		super();
		this.groupByColumnIndexs = groupByColumnIndexs;
		this.aggregateFunctionTypes = aggregateFunctionTypes;
		this.groupByFilter = groupByFilter;
	}

	public List<RowDataPacket> getRowDataPacketList() {
		if (!isMergAvg) {
			for (RowDataPacket rowDataPacket : rowDataPacketList) {
				mergAvg(rowDataPacket);
			}
			isMergAvg = true;
		}

		if (groupByFilter != null) {
			filterHaving();
		}

		return rowDataPacketList;
	}

	private void filterHaving() {
		if (groupByFilter.getColum() == null || rowDataPacketList == null) {
			return;
		}
		Iterator<RowDataPacket> iterator = rowDataPacketList.iterator();
		byte[] rightColumBytes = groupByFilter.getRight().getBytes(StandardCharsets.UTF_8);
		int columIndex = groupByFilter.getColum().getColumIndex();
		int columType = groupByFilter.getColum().getColumType();
		while (iterator.hasNext()) {
			RowDataPacket rowDataPacket = iterator.next();
			switch (groupByFilter.getOperator()) {
			case "=":
				if (notEq(rowDataPacket.fieldBytesList.get(columIndex), rightColumBytes, columType)) {
					iterator.remove();
				}
				break;
			case ">":
				if (notGt(rowDataPacket.fieldBytesList.get(columIndex), rightColumBytes, columType)) {
					iterator.remove();
				}
				break;
			case "<":
				if (notLt(rowDataPacket.fieldBytesList.get(columIndex), rightColumBytes, columType)) {
					iterator.remove();
				}
				break;
			case ">=":
				if (notGt(rowDataPacket.fieldBytesList.get(columIndex), rightColumBytes, columType) && notEq(rowDataPacket.fieldBytesList.get(columIndex), rightColumBytes, columType)) {
					iterator.remove();
				}
				break;
			case "<=":
				if (notLt(rowDataPacket.fieldBytesList.get(columIndex), rightColumBytes, columType) && notEq(rowDataPacket.fieldBytesList.get(columIndex), rightColumBytes, columType)) {
					iterator.remove();
				}
				break;
			case "!=":
				if (eq(rowDataPacket.fieldBytesList.get(columIndex), rightColumBytes, columType)) {
					iterator.remove();
				}
				break;
			}
		}

	}

	private boolean notLt(byte[] $leftBytes, byte[] $rightBytes, final int columType) {
		return -1 != compareObject($leftBytes, $rightBytes, columType);
	}

	private boolean notGt(byte[] $leftBytes, byte[] $rightBytes, final int columType) {
		return 1 != compareObject($leftBytes, $rightBytes, columType);
	}

	private boolean notEq(byte[] $leftBytes, byte[] $rightBytes, final int columType) {
		return 0 != compareObject($leftBytes, $rightBytes, columType);
	}

	private boolean eq(byte[] $leftBytes, byte[] $rightBytes, final int columType) {
		return 0 == compareObject($leftBytes, $rightBytes, columType);
	}

	public static final int compareObject(byte[] $leftBytes, byte[] $rightBytes, final int colType) {
		switch (colType) {
		case Column.COL_TYPE_SHORT:
		case Column.COL_TYPE_INT:
		case Column.COL_TYPE_INT24:
		case Column.COL_TYPE_LONG:
			return CompareUtil.compareInt(ByteUtil.getInt($leftBytes), ByteUtil.getInt($rightBytes));
		case Column.COL_TYPE_LONGLONG:
			return CompareUtil.compareLong(ByteUtil.getLong($leftBytes), ByteUtil.getLong($rightBytes));
		case Column.COL_TYPE_FLOAT:
		case Column.COL_TYPE_DOUBLE:
		case Column.COL_TYPE_DECIMAL:
		case Column.COL_TYPE_NEWDECIMAL:
			return CompareUtil.compareDouble(ByteUtil.getDouble($leftBytes), ByteUtil.getDouble($rightBytes));
		case Column.COL_TYPE_DATE:
		case Column.COL_TYPE_TIMSTAMP:
		case Column.COL_TYPE_TIME:
		case Column.COL_TYPE_YEAR:
		case Column.COL_TYPE_DATETIME:
		case Column.COL_TYPE_NEWDATE:
		case Column.COL_TYPE_BIT:
		case Column.COL_TYPE_VAR_STRING:
		case Column.COL_TYPE_STRING:
			// ENUM和SET类型都是字符串，按字符串处理
		case Column.COL_TYPE_ENUM:
		case Column.COL_TYPE_SET:
			return ByteUtil.compareNumberByte($leftBytes, $rightBytes);
			// BLOB相关类型和GEOMETRY类型不支持排序，略掉
		}
		return 0;
	}

	public void addRowDataPacket(RowDataPacket rowDataPacket) {
		for (RowDataPacket $rowDataPacket : rowDataPacketList) {
			if (sameGroupByColum(rowDataPacket, $rowDataPacket)) {
				aggregateRowDataPacket($rowDataPacket, rowDataPacket);
				return;
			}
		}
		// not aggreated ,insert new
		rowDataPacketList.add(rowDataPacket);

	}

	private void aggregateRowDataPacket(RowDataPacket oldRowDataPacket, RowDataPacket newRowDataPacket) {
		if (aggregateFunctionTypes == null) {
			return;
		}
		for (AggregateFunctionType aggregateFunctionType : aggregateFunctionTypes) {
			if (aggregateFunctionType.mergeType != AggregateFunctionType.AVG) {
				/**
				 * 进行非avg聚合处理
				 */
				byte[] result = mertFields(oldRowDataPacket.fieldBytesList.get(aggregateFunctionType.colum.columIndex), newRowDataPacket.fieldBytesList.get(aggregateFunctionType.colum.columIndex), aggregateFunctionType.colum.columType, aggregateFunctionType.mergeType);
				if (result != null) {
					oldRowDataPacket.fieldBytesList.set(aggregateFunctionType.colum.columIndex, result);
				}
			}
		}

	}

	private void mergAvg(RowDataPacket rowDataPacket) {
		if (aggregateFunctionTypes == null) {
			return;
		}

		for (AggregateFunctionType mergeColum : aggregateFunctionTypes) {
			if (mergeColum.mergeType == AggregateFunctionType.AVG) {
				byte[] result = mertFields(rowDataPacket.fieldBytesList.get(mergeColum.colum.avgSumIndex), rowDataPacket.fieldBytesList.get(mergeColum.colum.avgCountIndex), mergeColum.colum.columType, mergeColum.mergeType);
				if (result != null) {
					rowDataPacket.fieldBytesList.set(mergeColum.colum.avgSumIndex, result);
					rowDataPacket.fieldBytesList.remove(mergeColum.colum.avgCountIndex);
					rowDataPacket.fieldCount = rowDataPacket.fieldCount - 1;
				}
			}
		}

	}

	private byte[] mertFields(byte[] fieldValue1, byte[] fieldValue2, int columType, int mergeType) {
		if (fieldValue2 == null || fieldValue2.length == 0) {
			return fieldValue1;
		} else if (fieldValue1 == null || fieldValue1.length == 0) {
			return fieldValue2;
		}
		switch (mergeType) {
		/**
		 * sum相加
		 */
		case AggregateFunctionType.SUM:
			if (columType == Column.COL_TYPE_NEWDECIMAL || columType == Column.COL_TYPE_DOUBLE || columType == Column.COL_TYPE_FLOAT || columType == Column.COL_TYPE_DECIMAL) {
				Double value = ByteUtil.getDouble(fieldValue1) + ByteUtil.getDouble(fieldValue2);
				return value.toString().getBytes();
			}
			// continue to count case
		case AggregateFunctionType.COUNT: {
			long value = Long.parseLong(new String(fieldValue1)) + Long.parseLong(new String(fieldValue2));
			return LongUtil.toBytes(value);
		}
		case AggregateFunctionType.MAX: {
			int result = ByteUtil.compareNumberByte(fieldValue1, fieldValue2);
			return (result > 0) ? fieldValue1 : fieldValue2;

		}
		case AggregateFunctionType.MIN: {
			int result = ByteUtil.compareNumberByte(fieldValue1, fieldValue2);
			return (result > 0) ? fieldValue2 : fieldValue1;
		}
		case AggregateFunctionType.AVG: {
			double $double = ByteUtil.getDouble(fieldValue1);
			long $long = Long.parseLong(new String(fieldValue2));
			Double value = $double / $long;
			return value.toString().getBytes();
		}
		default:
			return null;
		}

	}

	private boolean sameGroupByColum(RowDataPacket newRow, RowDataPacket existRow) {
		if (groupByColumnIndexs == null) {// select count(*) from aaa , or group
			return true;
		}
		for (int i = 0; i < groupByColumnIndexs.length; i++) {
			if (!Arrays.equals(newRow.fieldBytesList.get(groupByColumnIndexs[i]), existRow.fieldBytesList.get(groupByColumnIndexs[i]))) {
				return false;
			}
		}
		return true;

	}
}
